package com.jiunuan.live.user.provider.config;

import com.alibaba.fastjson.JSON;
import com.jiunuan.live.common.interfaces.topic.UserProviderTopicNames;
import com.jiunuan.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import com.jiunuan.live.user.constants.CacheAsyncDeleteCode;
import com.jiunuan.live.user.dto.UserCacheAsyncDeleteDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;


import java.util.List;


@Configuration
@Slf4j
public class RocketMQConsumerConfig implements InitializingBean {

    @Resource
    private RocketMQConsumerProperties rocketMQConsumerProperties;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private UserProviderCacheKeyBuilder cacheKeyBuilder;
    @Override
    public void afterPropertiesSet() {
        initConsumer();
    }

    private void initConsumer() {
        try {
            //初始化rockerMQ消费者
            DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
            defaultMQPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameSrv());
            defaultMQPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupName());
            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            defaultMQPushConsumer.subscribe(UserProviderTopicNames.CACHE_ASYNC_DELETE_TOPIC,"*");
            // 设置消息监听器，用于处理接收到的消息
            defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently() {
                /**
                 * 处理消息的消费逻辑
                 * @param msgs 消息列表，包含需要消费的消息
                 * @param consumeConcurrentlyContext 消费消息的上下文环境
                 * @return 消息的消费状态
                 */
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    // 提取消息内容并转换为字符串
                    String json = new String(msgs.get(0).getBody());
                    // 解析消息内容为UserCacheAsyncDeleteDTO对象
                    UserCacheAsyncDeleteDTO userCacheAsyncDeleteDTO = JSON.parseObject(json, UserCacheAsyncDeleteDTO.class);
                    // 根据不同的缓存异步删除码处理消息
                    if (CacheAsyncDeleteCode.USER_INFO_DELETE.getCode() == userCacheAsyncDeleteDTO.getCode()) {
                        // 解析JSON获取用户ID，并删除对应的用户信息缓存
                        Long userId = JSON.parseObject(userCacheAsyncDeleteDTO.getJson()).getLong("userId");
                        redisTemplate.delete(cacheKeyBuilder.buildUserInfoKey(userId));
                        log.info("延迟删除用户信息缓存，userId is {}",userId);
                    } else if (CacheAsyncDeleteCode.USER_TAG_DELETE.getCode() == userCacheAsyncDeleteDTO.getCode()) {
                        // 解析JSON获取用户ID，并删除对应的用户标签缓存
                        Long userId = JSON.parseObject(userCacheAsyncDeleteDTO.getJson()).getLong("userId");
                        redisTemplate.delete(cacheKeyBuilder.buildTagKey(userId));
                        log.info("延迟删除用户标签缓存，userId is {}",userId);
                    }
                    // 指示消息消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            defaultMQPushConsumer.start();
            log.info("RocketMQConsumer start success,namesrvAddr is {}", rocketMQConsumerProperties.getNameSrv());

        }catch (MQClientException e){
            throw new RuntimeException(e);
        }

    }
}
